package com.wudl.flink;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ibm.icu.impl.Row;
import com.sun.jersey.core.util.StringIgnoreCaseKeyComparator;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.Properties;

/**
 * @ClassName : FlinkCdcWithCustomer
 * @Description :
 * @Author :wudl
 * @Date: 2021-03-13 23:37
 */

public class FlinkCdcWithCustomer {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties proper = new Properties();
        proper.setProperty("scan.startup.mode", "initial");
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("192.168.1.153")
                .username("root")
                .password("slave012021@123ABM")
                .port(3306)
                .databaseList("test")
//                .tableList("test.test02")
                .deserializer(new MySchema())
                .debeziumProperties(proper)
                .build();
        DataStreamSource<String> dataStreamSource = env.addSource(mysqlSource);
        dataStreamSource.print();
        SingleOutputStreamOperator<String> streamOperator = dataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String typeStr = (String) jsonObject.get("type");

                if (typeStr.equals("select")  )
                {
                    return  null;
                }
                String sql = Utils.opreator(s);
//                String sql = "DELETE FROM activity WHERE id = 1";
                System.out.println("sql--"+sql);
                return sql;
            }
        });

        streamOperator.addSink(new SinkToMySQL());

//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.executeSql(" INSERT  INTO `activity`(`id`,`activity_name`,`activity_type`,`activity_desc`,`start_time`,`end_time`,`create_time`) VALUES (1,'联想品牌优惠活动','2201','满减活动 ','2020-02-27 17:11:09','2020-02-29 17:11:12','2020-02-27 17:11:16') "  +
//                "WITH ( " +
//                " 'connector' = 'mysql-cdc', " +
//                " 'hostname' = '192.168.1.180', " +
//                " 'port' = '3306', " +
//                " 'username' = 'root', " +
//                " 'password' = '123456', " +
//                " 'database-name' = 'test', " +
//                " 'table-name' = 'activity' " +
//                ")");
//
//        //3.打印
//        Table table = tableEnv.sqlQuery("select * from test.activity");
//        tableEnv.toRetractStream(table, Row.class).print();


        env.execute();


    }


    private static class MySchema implements DebeziumDeserializationSchema<String> {

        /*
         * 反序列化方法
         */
        @Override
        public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {

            String topic = sourceRecord.topic();
            String[] split = topic.split("\\.");
            String dbName = split[1];
            String table = split[2];

            // 获取数据
            Struct value = (Struct) sourceRecord.value();
            Struct after = value.getStruct("after");
            Struct before = value.getStruct("before");
            JSONObject data = new JSONObject();
            if (after != null) {
                Schema schema = after.schema();
                for (Field field : schema.fields()) {
                    if (field.name().contains("time"))
                    {
                        String formatDate = DateUtils.getFormatDate(after.get(field.name()).toString());
                        System.out.println("---field.name()"+field.name());
                        data.put(field.name(), formatDate);
                    }else {
                        data.put(field.name(), after.get(field.name()));
                    }
                }
            }
            if (before != null) {
                Schema schema = before.schema();
                for (Field field : schema.fields()) {
                    System.out.println("------------"+field.name()+"==========" +before.get(field.name()));
                    if (field.name().contains("time"))
                    {
                        String formatDate = DateUtils.getFormatDate(before.get(field.name()).toString());
                        System.out.println("---field.name()"+field.name());
                        data.put(field.name(), formatDate);
                    }else
                    {
                        data.put(field.name(), before.get(field.name()));
                    }

                }
            }

            // 获取操作类型
            Envelope.Operation operation = Envelope.operationFor(sourceRecord);

            // 创建json用于存最终的结果
            JSONObject result = new JSONObject();
            result.put("database", dbName);
            result.put("table", table);
            result.put("type", operation.toString().toLowerCase());
            result.put("date", data);

            collector.collect(result.toJSONString());
        }

        @Override
        public TypeInformation<String> getProducedType() {
            return TypeInformation.of(String.class);
        }
    }
}
